package com.etc

/**
  * @Title:
  * @ProjectName
  * @Description:
  * @author kalista
  */

import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.HasOffsetRanges
import org.apache.spark.streaming.{Seconds, StreamingContext}

object DirectKafkaMeterData {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(conf, Seconds(15))
    //kafka节点
    val BROKER_LIST = "master:9092,slave1:9092,slave2:9092"
    val ZK_SERVERS = "master:2181,slave1:2181,slave2:2181"
    val GROUP_ID = "test-consumer-group" //消费者组

    val topics = Set("test") //待消费topic

    /*
    参数说明
    AUTO_OFFSET_RESET_CONFIG
        smallest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
        largest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
        disable:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     */
    val kafkaParams = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> BROKER_LIST,
      ConsumerConfig.GROUP_ID_CONFIG -> GROUP_ID,
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "smallest"
    )
    val kafkaManager = new KafkaManager(kafkaParams)
    //创建数据流
    val kafkaStream: InputDStream[(String, String)] = kafkaManager.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,topics)

    kafkaStream.foreachRDD { rdd =>
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd.map(msg => msg._2).foreachPartition(ite=>{
        ite.foreach(record => {
          //处理数据的方法
          println(record)
        })
      })
      kafkaManager.updateZKOffsets(offsetRanges)
    }
    ssc.start()
    ssc.awaitTermination()
  }

}
